#ifndef XG_SYNC_CPP
#define XG_SYNC_CPP
///////////////////////////////////////////////////////////////////
#include "../File.h"
#include "../Reflect.h"

Object::~Object()
{
}
string Object::toString() const
{
	char buffer[16];

	sprintf(buffer, "%p", this);

	return buffer;
}
const char* Object::getClassName() const
{
#ifdef _MSC_VER
	return typeid(*this).name() + 6;
#else
	return SkipStartString(typeid(*this).name(), "0123456789");
#endif
}
string Object::attrs(const string& exclude) const
{
	StringCreator out;
	vector<ReflectItem> vec = ReflectHelper::GetAttrList(this);

	if (exclude.empty())
	{
		for (ReflectItem& item : vec)
		{
			out << "," << item.getName();
		}
	}
	else
	{
		char tag[1024];
		string exlist = "," + exclude + ",";

		for (ReflectItem& item : vec)
		{
			sprintf(tag, ",%s,", item.getName());

			if (exlist.find(tag) == string::npos)
			{
				out << "," << item.getName();
			}
		}
	}

	const string& res = out.getContent();

	return res.empty() ? res : res.substr(1);
}

void WorkItem::run()
{
}
bool WorkItem::runnable()
{
	return true;
}
void WorkItem::wait() const
{
	while (true)
	{
		if (isDone()) return;

		Sleep(1);
	}
}
bool WorkItem::wait(int timeout) const
{
	Timer timer(NULL);

	timeout *= 1000;

	while (true)
	{
		if (isDone()) return true;

		Sleep(1);

		if (timer.getTimeGap() >= timeout) return false;
	}
}

class Runnable : public WorkItem
{
protected:
	long long etime;
	function<void()> func;

public:
	void run()
	{
		func();
	}
	bool runnable()
	{
		return etime <= 0 || GetTime() > etime;
	}
	void setDelay(int delay)
	{
		etime = delay > 0 ? GetTime() + delay * 1000 : 0;
	}
	Runnable(function<void()> run)
	{
		func = run;
		etime = 0;
	}
	Runnable(function<void()> run, int delay)
	{
		func = run;
		setDelay(delay);
	}
};

sp<WorkItem> stdx::async(sp<WorkItem> item)
{
	static TaskQueue* queue = TaskQueue::Instance();

	while (true)
	{
		if (queue->push(item)) break;

		Sleep(1);
	}

	return item;
}
sp<WorkItem> stdx::async(function<void()> func)
{
	sp<WorkItem> item = newsp<Runnable>(func);

	return async(item);
}


bool Thread::start()
{
	CATCH_EXCEPTION({
		std::thread(std::bind(&Thread::run, this)).detach();

		return true;
	});

	return false;
}

void TaskQueue::run()
{
	++threadcount;

	while (threads > 0)
	{
		sp<WorkItem> item;
		size_t len = pop(item);

		if (len > 0)
		{
			bool errdone = true;

			CATCH_EXCEPTION({
				if (item->runnable())
				{
					item->run();
					item->done = XG_OK;
					--curdepth;
				}
				else
				{
					static WorkItem* last = NULL;

					if (last == item.get())
					{
						Sleep(1);
					}
					else
					{
						last == item.get();
					}

					while (true)
					{
						{
							SpinLocker lk(mtx);

							if (queue.push(item)) break;
						}

						Sleep(1);
					}
				}

				errdone = false;
			});

			if (errdone) item->done = XG_ERROR;
		}
		else
		{
			if (threadcount > threads)
			{
				SpinLocker lk(mtx);

				if (threadcount > threads)
				{
					removeDestroyFunc();
					--threadcount;

					return;
				}
			}

			Sleep(1);
		}
	}

	--threadcount;
}
bool TaskQueue::push(sp<WorkItem> item)
{
	CHECK_FALSE_RETURN(item);

	item->done = 0;

	{
		SpinLocker lk(mtx);
		size_t len = queue.push(item);

		if (len > 0)
		{
			size_t num = threads + threads;

			if (len > num && threadcount < num)
			{
				CATCH_EXCEPTION({
					std::thread(std::bind(&TaskQueue::run, this)).detach();
				});
			}

			++curdepth;

			return true;
		}
	}

	CHECK_FALSE_RETURN(threads == 0 && start());

	SpinLocker lk(mtx);

	CHECK_FALSE_RETURN(queue.push(item));

	++curdepth;

	return true;
}
void TaskQueue::cancelDestroyFunc()
{
	long key = GetCurrentThreadId();
	SpinLocker lk(desmtx);
	destroymap.erase(key);
}
bool TaskQueue::removeDestroyFunc()
{
	function<void()> func;
	long key = GetCurrentThreadId();

	{
		SpinLocker lk(desmtx);
		auto it = destroymap.find(key);

		if (it == destroymap.end()) return false;

		func = it->second;
		destroymap.erase(it);
	}

	func();

	return true;
}
bool TaskQueue::setDestroyFunc(function<void()> func)
{
	long key = GetCurrentThreadId();
	SpinLocker lk(desmtx);
	auto it = destroymap.find(key);

	CHECK_FALSE_RETURN(it == destroymap.end());

	destroymap[key] = func;

	return true;
}
bool TaskQueue::start(size_t threads, size_t maxsz)
{
	SpinLocker lk(mtx);

	if (queue.capacity() == 0) queue.init(maxsz);

	this->threads = threads;

	CATCH_EXCEPTION({
		while (threads-- > 0) std::thread(std::bind(&TaskQueue::run, this)).detach();

		return true;
	});

	return false;
}
TaskQueue* TaskQueue::Instance()
{
	XG_DEFINE_GLOBAL_VARIABLE(TaskQueue)
}

bool TimerTaskQueue::Task::update(long long now)
{
	if (++times >= maxtimes && maxtimes > 0) return false;

	if (delay > 0)
	{
		rtime = now + delay;
	}
	else
	{
		rtime = now - delay - now % (24 * 3600 * 1000);

		if (rtime <= now) rtime += 24 * 3600 * 1000;
	}

	return true;
}
bool TimerTaskQueue::get(Task& task)
{
	long long now = GetTime() / 1000;
	
	if (now <= 0) return false;

	SpinLocker lk(mtx);

	if (queue.empty()) return false;

	task = queue.top();

	if (now < task.rtime) return false;

	queue.pop();

	if (task.update(now)) queue.push(task);

	return true;
}

void TimerTaskQueue::run()
{
	Task item;
	TaskQueue* taskqueue = TaskQueue::Instance();

	while (true)
	{
		for (int i = 0; i < 10 && get(item); i++)
		{
			while (!taskqueue->push(item.task)) Sleep(1);
		}

		Sleep(1);
	}
}
void TimerTaskQueue::check(function<bool(sp<WorkItem>)> filter)
{
	SpinLocker lk(mtx);
	priority_queue<Task> tmp;

	while (!queue.empty())
	{
		if (filter(queue.top().task)) tmp.push(queue.top());

		queue.pop();
	}

	std::swap(tmp, queue);
}
bool TimerTaskQueue::push(sp<WorkItem> task, int delay, int maxtimes)
{
	Task item;

	CHECK_FALSE_RETURN(item.init(delay, maxtimes, task));

	mtx.lock();

	if (started == 0)
	{
		std::thread(std::bind(&TimerTaskQueue::run, this)).detach();

		started = XG_OK;
	}

	queue.push(item);

	mtx.unlock();

	return true;
}
bool TimerTaskQueue::daily(sp<WorkItem> task, const string& time, int maxtimes)
{
	DateTime dt = DateTime::FromString(DateTime::ToString().substr(0, 11) + time);

	CHECK_FALSE_RETURN(dt.canUse());

	int delay = (dt.getTime() * 1000) % (24 * 3600 * 1000);

	return push(task, -delay, maxtimes);
}
bool TimerTaskQueue::update(int delay, int maxtimes, function<bool(sp<WorkItem>)> filter)
{
	int res = 0;
	SpinLocker lk(mtx);
	priority_queue<Task> tmp;

	while (!queue.empty())
	{
		Task item = queue.top();

		if (filter(item.task))
		{
			item.init(delay, maxtimes, item.task);
			res++;
		}

		tmp.push(item);
		queue.pop();
	}

	std::swap(tmp, queue);

	return res > 0;
}

TimerTaskQueue* TimerTaskQueue::Instance()
{
	XG_DEFINE_GLOBAL_VARIABLE(TimerTaskQueue)
}

bool stdx::parallel(vector<function<void()>> items)
{
	size_t count = items.size();

	if (count <= 0) return true;

	vector<sp<WorkItem>> vec;

	auto push = [&](const function<void()>& func){
		while (true)
		{
			auto item = async(func);
			
			if (item)
			{
				vec.push_back(item);

				break;
			}

			Sleep(1);
		}
	};

	for (size_t i = 1; i < count; i++)
	{
		push(items[i]);
	}

	bool errdone = true;

	CATCH_EXCEPTION({
		items[0]();

		errdone = false;
	});

	for (auto& item : vec)
	{
		item->wait();

		if (item->isErrorDone()) errdone = true;
	}

	return errdone ? false : true;
}

void stdx::timer(int delay, sp<WorkItem> item)
{
	stdx::delay(delay, item, 0);
}
void stdx::timer(int delay, function<void()> func)
{
	stdx::delay(delay, func, 0);
}
void stdx::delay(int delay, sp<WorkItem> item, int maxtimes)
{
	static TimerTaskQueue* queue = TimerTaskQueue::Instance();

	assert(delay > 0);

	while (true)
	{
		if (queue->push(item, delay, maxtimes)) break;

		Sleep(1);
	}
}
void stdx::delay(int delay, function<void()> func, int maxtimes)
{
	sp<WorkItem> item = newsp<Runnable>(func);

	stdx::delay(delay, item, maxtimes);
}

///////////////////////////////////////////////////////////////////
#endif